package com.gengzp.mq.server.service.impl;

import com.alibaba.nacos.shaded.com.google.common.collect.Lists;
import com.fasterxml.jackson.core.type.TypeReference;
import com.gengzp.mq.common.model.base.MqPaginationResult;
import com.gengzp.mq.common.model.dto.GengzpMqConsumerDto;
import com.gengzp.mq.common.model.dto.SearchGengzpMqConsumerDto;
import com.gengzp.mq.common.model.entity.GengzpMqMessage;
import com.gengzp.mq.common.model.entity.GengzpMqTopic;
import com.gengzp.mq.common.model.req.AddGengzpMqTopicReq;
import com.gengzp.mq.common.model.req.EditGengzpMqTopicReq;
import com.gengzp.mq.common.model.req.PageGengzpMqTopicReq;
import com.gengzp.mq.common.model.resp.PageGengzpMqTopicResp;
import com.gengzp.mq.common.model.resp.SearchGengzpMqConsumerResp;
import com.gengzp.mq.common.utils.GengzpMqAssertUtils;
import com.gengzp.mq.common.utils.GengzpMqJsonUtils;
import com.gengzp.mq.common.utils.GengzpMqRedisUtils;
import com.gengzp.mq.common.utils.GengzpMqSnowflakeIdUtils;
import com.gengzp.mq.server.facade.IGengzpMqFacadeService;
import com.gengzp.mq.server.mapper.GengzpMqTopicMapper;
import com.gengzp.mq.server.service.IGengzpMqTopicService;
import com.google.common.collect.Maps;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static com.gengzp.mq.common.constants.GengzpMqCacheConstants.TOPIC_CONSUMER_SERVICE_NAME_KEY_PREFIX;
import static com.gengzp.mq.common.constants.GengzpMqCacheConstants.TOPIC_INFO_KEY_PREFIX;

/**
 * @ClassName GengzpMqTopicServiceImpl
 * @Description 消息队列模块主题管理服务层实现
 * @Author gengzp
 * @Date 2025/7/10 10:13
 */
@Service
public class GengzpMqTopicServiceImpl implements IGengzpMqTopicService {

    @Autowired
    private GengzpMqTopicMapper gengzpMqTopicMapper;

    @Autowired
    private IGengzpMqFacadeService gengzpMqFacadeService;

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void addTopic(AddGengzpMqTopicReq req) {
        if (Objects.nonNull(gengzpMqTopicMapper.selectTopicByName(req.getTopic()))) {
            throw new RuntimeException(String.format("该主题: [%s] 已存在, 添加失败", req.getTopic()));
        }

        GengzpMqTopic mqTopic = new GengzpMqTopic();
        mqTopic.setTopic(req.getTopic());
        mqTopic.setDescription(req.getDescription());

        // 主键id
        mqTopic.setId(GengzpMqSnowflakeIdUtils.nextId());
        // 当前系统时间
        Date date = new Date();
        mqTopic.setCreatedTime(date);
        mqTopic.setLastModifiedTime(date);
        gengzpMqTopicMapper.insertTopic(mqTopic);

        // 缓存到redis
        GengzpMqRedisUtils.set(TOPIC_INFO_KEY_PREFIX + req.getTopic(),
                GengzpMqJsonUtils.toJson(mqTopic), 1, TimeUnit.DAYS);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void editTopic(EditGengzpMqTopicReq req) {
        GengzpMqTopic gengzpMqTopic = gengzpMqTopicMapper.selectTopicById(req.getId());
        GengzpMqAssertUtils.nonNull(gengzpMqTopic, "该主题不存在, 修改失败");

        gengzpMqTopic.setDescription(req.getDescription());

        // 修改时间
        gengzpMqTopic.setLastModifiedTime(new Date());
        gengzpMqTopicMapper.updateTopicById(gengzpMqTopic);

        // 缓存到redis
        GengzpMqRedisUtils.set(TOPIC_INFO_KEY_PREFIX + gengzpMqTopic.getTopic(),
                GengzpMqJsonUtils.toJson(gengzpMqTopic), 1, TimeUnit.DAYS);
    }

    @Override
    public void deleteTopicById(Long topicId) {
        gengzpMqTopicMapper.deleteTopicById(topicId);
    }

    @Override
    public MqPaginationResult<PageGengzpMqTopicResp> pageQueryTopic(PageGengzpMqTopicReq req) {
        // 分页查询系统用户
        List<GengzpMqTopic> mqTopics = gengzpMqTopicMapper.pageQueryGengzpMqTopic(
                (req.getPageNum() - 1) * req.getPageSize(),
                req.getPageSize(),
                req.getTopic(),
                req.getDescription());
        if (CollectionUtils.isEmpty(mqTopics)) {
            return MqPaginationResult.get(req.getPageNum(), req.getPageSize(), 0, Collections.emptyList());
        }

        // 根据查询到的主题名, 查询所有消息
        List<GengzpMqMessage> gengzpMqMessages = gengzpMqFacadeService.searchMessageByTopics(
                mqTopics.stream().map(GengzpMqTopic::getTopic).collect(Collectors.toList()));

        // 主题对应消息集合映射   key:主题名  value:消息集合
        Map<String, List<GengzpMqMessage>> topic2MessageMap = Maps.newHashMap();
        if (CollectionUtils.isNotEmpty(gengzpMqMessages)) {
            topic2MessageMap = gengzpMqMessages.stream().collect(Collectors.groupingBy(GengzpMqMessage::getTopic));
        }

        // 组装返回数据
        List<PageGengzpMqTopicResp> result = Lists.newArrayList();
        for (GengzpMqTopic mqTopic : mqTopics) {
            PageGengzpMqTopicResp resp = new PageGengzpMqTopicResp();
            resp.setId(String.valueOf(mqTopic.getId()));
            resp.setTopic(mqTopic.getTopic());
            resp.setDescription(mqTopic.getDescription());
            resp.setCreatedTime(mqTopic.getCreatedTime());
            resp.setLastModifiedTime(mqTopic.getLastModifiedTime());

            // 取出当前主题下的所有消息
            List<GengzpMqMessage> topicRelationMessages = topic2MessageMap.getOrDefault(mqTopic.getTopic(), Collections.emptyList());
            resp.setTotalMessageCount(topicRelationMessages.size());
            resp.setUnConsumedMessageCount(0);
            resp.setConsumingMessageCount(0);
            resp.setConsumedMessageCount(0);
            resp.setConsumedFailMessageCount(0);
            if (CollectionUtils.isNotEmpty(topicRelationMessages)) {
                for (GengzpMqMessage message : topicRelationMessages) {
                    switch (message.getStatus()) {
                        case 0: // 未消费
                            resp.setUnConsumedMessageCount(resp.getUnConsumedMessageCount() + 1);
                            break;
                        case 1: // 消费中
                            resp.setConsumingMessageCount(resp.getConsumingMessageCount() + 1);
                            break;
                        case 2: // 已消费
                            resp.setConsumedMessageCount(resp.getConsumedMessageCount() + 1);
                            break;
                        case 3: // 消费失败
                            resp.setConsumedFailMessageCount(resp.getConsumedFailMessageCount() + 1);
                            break;
                    }
                }
            }

            result.add(resp);
        }

        return MqPaginationResult.get(
                req.getPageNum(),
                req.getPageSize(),
                gengzpMqTopicMapper.selectCountGengzpMqTopic(req.getTopic(), req.getDescription()),
                result);
    }

    @Override
    public List<String> searchTopicNames() {
        return gengzpMqTopicMapper.selectTopicNames();
    }

    @Override
    public SearchGengzpMqConsumerResp searchConsumer(String topic) {
        SearchGengzpMqConsumerResp result = new SearchGengzpMqConsumerResp();
        result.setTopic(topic);
        result.setConsumers(Lists.newArrayList());

        // 从redis查询出当前topic的所有消费者服务信息
        Object topicRelationConsumerObj = GengzpMqRedisUtils.get(TOPIC_CONSUMER_SERVICE_NAME_KEY_PREFIX + topic);
        if (Objects.isNull(topicRelationConsumerObj)) {
            return result;
        }

        Date currentDate = new Date();
        List<GengzpMqConsumerDto> topicRelationConsumers = GengzpMqJsonUtils.parseObjectWithGeneric(topicRelationConsumerObj.toString(), new TypeReference<>() {
        });
        for (GengzpMqConsumerDto topicRelationConsumer : topicRelationConsumers) {
            SearchGengzpMqConsumerDto resultDto = new SearchGengzpMqConsumerDto();
            resultDto.setConsumerServiceName(topicRelationConsumer.getConsumerServiceName());
            resultDto.setConsumerServiceIp(topicRelationConsumer.getConsumerServiceIp());
            resultDto.setSubscribeStatus(topicRelationConsumer.getSubscribeEndTime().before(currentDate) ? "已断开" : "订阅中");
            result.getConsumers().add(resultDto);
        }

        return result;
    }

}
